package com.imooc.bigdata.chapter08_my.business

import com.imooc.bigdata.chapter08.utils.{KuduUtils, SQLUtils, SchemaUtils}
import com.imooc.bigdata.chapter08_my.SparkApp.logError
import com.imooc.bigdata.chapter08_my.`trait`.DataProcess
import com.imooc.bigdata.chapter08_my.utils.{DateUtils, IPUtils}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

object LogETLProcess extends DataProcess{
  override def process(spark: SparkSession): Unit = {

    import spark.implicits._

    //这是外面传入的参数 spark.raw.path
    val rowPath: String = spark.sparkContext.getConf.get("spark.raw.path")

    var jsonDF: DataFrame = null
      if (null != rowPath && ""!= rowPath && rowPath.length!= 0){
        jsonDF = spark.read.json(rowPath)
    }else{
      System.out.println("rwSource文件路径为空")
      System.exit(0)
    }

    //    //加载要处理的数据源(Data Source)，这里是本地
//    var jsonDF: DataFrame = spark.read.json("E:\\IntelliJ-IDEA\\sparksql-train\\data\\data-test.json")

    /*
       1  这里是网上的ip地址库，然后找出对应的地区
       2  也可以用gitHub 上面的 纯真的ip解析
       3  但是生产的时候，用专门的公司，付钱的ip服务
     */
    //加载本地ip库

    val ipRulePath: String = spark.sparkContext.getConf.get("spark.ip.path")

    var ipRdd: RDD[String] = null
      if (null != ipRulePath && ""!= ipRulePath && ipRulePath.length!= 0){
      ipRdd= spark.sparkContext
        .textFile(ipRulePath)
    }else{
      System.out.println("ipSource文件路径为空")
      System.exit(0)
    }


    //把这个RDD读取的数据拆分出需要的字段，然后转成DF，注册成一张规则表
    val ipRuleDF: DataFrame = ipRdd.map(x => {
      val splits = x.split("\\|") //解析出这个地址库的前几个字段用
      val startIP: Long = splits(2).toLong
      val endIP: Long = splits(3).toLong
      val province: String = splits(6)
      val city: String = splits(7)
      val isp: String = splits(9)
      (startIP, endIP, province, city, isp)
    }).toDF("start_ip", "end_ip", "province", "city", "isp")

    /*
     将数据中的ip和ipRuleDF中的ip对应，得到相应的省份，城市，网络运营商。
     现在有两个DF，一个是原始数据DF 一个是规则化DF, 合并用join操作， 条件是ipRdd的ip得在ipRuleDF ip 范围内

    但是json里面的ip是字符串，192.168.2.19  ip地址库里面的是3059465949  Long类型
    所以下面需要转换需要转换
     */

    //自定义函数， 调用ip转换类
    import org.apache.spark.sql.functions._
    def getLongIP() = udf((ip:String)=>{
      IPUtils.ip2Long(ip)
    })

    //替换jsonDF里面的ip为转换了的 ip_Long
    jsonDF = jsonDF.withColumn("ip_Long",getLongIP()($"ip"))

    //    //API方式
    //    jsonDF.join(ipRuleDF,jsonDF("ip_Long")
    //      .between(ipRuleDF("start_ip"),ipRuleDF("end_ip")))
    //        .show()

    //SQL方式
    jsonDF.createOrReplaceTempView("logs")
    ipRuleDF.createOrReplaceTempView("ips")

    val sql = SQLUtils.SQL
    val dataFrame: DataFrame = spark.sql(sql)
    val MASTER = "spark000"
    val TABLE_NAME = DateUtils.getTableName("ods",spark) //这样返回 name+time 的新名字
    val partitionID = "ip"
    val schema = SchemaUtils.ODSSchema

    KuduUtils.sink(dataFrame,TABLE_NAME,MASTER,schema,partitionID)

  }
}
